package com.atguigu.chapter12;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/7/26 13:47
 */
public class Flink01_TopN {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        
        // 1. 建立一个表与source关联
        tenv.executeSql("create table ub(" +
                            "   user_id bigint, " +
                            "   item_id bigint, " +
                            "   category_id int, " +
                            "   behavior string, " +
                            "   ts bigint," +
                            "   et as to_timestamp(from_unixtime(ts)), " +
                            "   watermark for et as et - interval '5' second" +
                            ")with(" +
                            "   'connector' = 'filesystem', " +
                            "   'path' = 'input/UserBehavior.csv', " +
                            "   'format' = 'csv' " +
                            ")");
        
        //tenv.sqlQuery("select * from ub").execute().print();
        // 2. 开窗聚合: 计算每个商品在每个窗口内的点击量
        Table t1 = tenv.sqlQuery("select" +
                                     " item_id," +
                                     " hop_start(et, interval '10' minute, interval '1' hour) w_start," +
                                     " hop_end(et, interval '10' minute, interval '1' hour) w_end," +
                                     " count(*) item_count " +
                                     "from ub " +
                                     "where behavior = 'pv' " +
                                     "group by item_id, hop(et, interval '10' minute, interval '1' hour)");
        tenv.createTemporaryView("t1", t1);
        
        // 3. 使用over窗口, 按照点击量排序, 给每个点击量配置一个名次
        Table t2 = tenv.sqlQuery("select" +
                                     "   *,  " +
                                     "   row_number() over(partition by w_end order by item_count desc) rk " +
                                     " from t1");
        
        tenv.createTemporaryView("t2", t2);
        // 4. 过滤出来名次小于等于3的
        Table t3 = tenv.sqlQuery("select " +
                                     "  item_id," +
                                     "  w_end," +
                                     "  item_count," +
                                     "  rk " +
                                     "from t2  " +
                                     "where rk<=3");
        
        // 5. 数据写入到mysql中
        // 5.1 创建动态表与mysql进行关联
        tenv.executeSql("create table hot_item(" +
                            "   item_id bigint, " +
                            "   w_end timestamp(3), " +
                            "   item_count bigint, " +
                            "   rk bigint, " +
                            "   PRIMARY KEY (w_end, rk) NOT ENFORCED" +
                            ")with(" +
                            "   'connector' = 'jdbc', " +
                            "   'url' = 'jdbc:mysql://hadoop162:3306/flink_sql?useSSL=false', " +
                            "   'table-name' = 'hot_item', " +
                            "   'username' = 'root', " +
                            "   'password' = 'aaaaaa' " +
                            ")"
        );
        
        t3.executeInsert("hot_item");
        
    }
}
